import { SpanKind } from "@opentelemetry/api";
import {
  Job,
  type JobsOptions,
  Queue,
  type QueueOptions,
  type RedisClient,
} from "bullmq";
import { EventEmitter } from "events";
import { getLangWatchTracer } from "langwatch";
import { createLogger } from "../../../utils/logger";
import { connection } from "../../redis";

const logger = createLogger("langwatch:queueWithFallback");

// Queue that falls back to calling the worker directly if the queue is not available
export class QueueWithFallback<
  DataType,
  ResultType,
  NameType extends string,
> extends Queue<
  DataType,
  ResultType,
  NameType,
  DataType,
  ResultType,
  NameType
> {
  private worker: (job: Job<DataType, ResultType, NameType>) => Promise<any>;
  private tracer = getLangWatchTracer("langwatch.queueWithFallback");

  constructor(
    name: string,
    worker: (job: Job<DataType, ResultType, NameType>) => Promise<any>,
    opts?: QueueOptions,
  ) {
    super(name, opts, connection ? undefined : (NoOpConnection as any));
    this.worker = worker;
  }

  async add(
    name: NameType,
    data: DataType,
    opts?: JobsOptions,
  ): Promise<Job<DataType, ResultType, NameType>> {
    return await this.tracer.withActiveSpan(
      `FallbackQueue${this.name}.add`,
      {
        kind: SpanKind.INTERNAL,
        attributes: {
          "queue.name": name,
          "queue.id": opts?.jobId,
        },
      },
      async () => {
        if (!connection) {
          await new Promise((resolve) => setTimeout(resolve, opts?.delay ?? 0));
          await this.worker(new Job(this, name, data, opts));
        }

        try {
          const timeoutState = { state: "waiting" };
          const timeoutPromise = new Promise((resolve, reject) => {
            setTimeout(() => {
              if (timeoutState.state === "waiting") {
                reject(
                  new Error(
                    `Timed out after 3s trying to insert on the queue ${this.name}`,
                  ),
                );
              } else {
                resolve(undefined);
              }
            }, 3000);
          });

          const job = await Promise.race([
            timeoutPromise,
            super.add(name, data, opts).then((job) => {
              timeoutState.state = "resolved";
              return job;
            }),
          ]);

          return job as Job<DataType, ResultType, NameType>;
        } catch (error) {
          logger.error(
            { error },
            `failed sending to redis ${this.name} inserting trace directly, attempting to process job synchronously`,
          );

          return await this.worker(new Job(this, name, data, opts));
        }
      },
    );
  }

  async addBulk(
    jobs: { name: NameType; data: DataType; opts?: JobsOptions }[],
  ): Promise<Job<DataType, ResultType, NameType>[]> {
    return await this.tracer.withActiveSpan(
      `FallbackQueue${this.name}.addBulk`,
      {
        kind: SpanKind.INTERNAL,
        attributes: {
          "queue.name": jobs.map((job) => job.name).join(","),
        },
      },
      async () => {
        if (!connection) {
          await Promise.all(
            jobs.map(async (job) => this.add(job.name, job.data, job.opts)),
          );
        }
        return await super.addBulk(jobs);
      },
    );
  }

  // @ts-ignore
  async getJob(
    id: string,
  ): Promise<Job<DataType, ResultType, NameType> | undefined> {
    return await this.tracer.withActiveSpan(
      `FallbackQueue${this.name}.getJob`,
      {
        kind: SpanKind.INTERNAL,
        attributes: {
          "queue.name": id,
        },
      },
      async () => {
        if (!connection) {
          return undefined;
        }

        const timeoutState = { state: "waiting" };
        const timeoutPromise = new Promise((resolve, reject) => {
          setTimeout(() => {
            if (timeoutState.state === "waiting") {
              reject(
                new Error("Timed out after 3s trying to get job from redis"),
              );
            } else {
              resolve(undefined);
            }
          }, 3000);
        });

        try {
          const job = await Promise.race([
            timeoutPromise,
            super.getJob(id).then((job) => {
              timeoutState.state = "resolved";
              return job;
            }),
          ]);

          return job as Job<DataType, ResultType, NameType>;
        } catch (error) {
          logger.error({ error }, "failed getting job from redis");
          return undefined;
        }
      },
    );
  }
}

class NoOpConnection extends EventEmitter {
  constructor() {
    super();
  }
  async reconnect(): Promise<void> {
    return void 0;
  }
  async disconnect(): Promise<void> {
    return void 0;
  }
  async waitUntilReady(): Promise<void> {
    return void 0;
  }
  async close(): Promise<void> {
    return void 0;
  }
  get client(): Promise<RedisClient> {
    return Promise.resolve(null as any);
  }
}
